package cn.com.basic.framework.datahub;

import cn.com.basic.framework.datahub.error.PublishError;
import cn.com.basic.framework.exception.exceptions.SysException;
import com.aliyun.datahub.client.DatahubClient;
import com.aliyun.datahub.client.model.BlobRecordData;
import com.aliyun.datahub.client.model.PutRecordsResult;
import com.aliyun.datahub.client.model.RecordEntry;
import com.aliyun.datahub.client.model.TupleRecordData;
import com.aliyun.datahub.common.util.RetryUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * 发布者，用于上传实时数据的类
 *
 * @author zhuxj
 */
public class DatahubPublisher {

    private static Logger logger = LoggerFactory.getLogger(DatahubPublisher.class);

    private String projectName;             //项目空间
    private String topicName;               //topic名称
    private DatahubClient client;           //datahub客户端实例
    private int retryTimes = 1;             //写入重试次数
    private long sleepTime = 2000;          //写入重试间隔
    private boolean exponential = false;    //间隔是否指数增长

    /**
     * 唯一构造
     *
     * @param projectName 项目空间名称
     * @param topicName   主题名称
     */
    public DatahubPublisher(DatahubClient client, String projectName, String topicName) {
        this.projectName = projectName;
        this.topicName = topicName;
        this.client = client;

    }

    /**
     * 写入Tuple表记录
     *
     * @param recordData List of {@link TupleRecordData} 待写入的记录列表
     */
    public void writeTuple(List<TupleRecordData> recordData) {

        List<RecordEntry> recordEntries = new ArrayList<>();

        for (TupleRecordData recordDatum : recordData) {
            RecordEntry recordEntry = new RecordEntry();
            recordEntry.setRecordData(recordDatum);
            recordEntries.add(recordEntry);
        }
        try {
            PutRecordsResult recordsResult = commitRecordEntries(recordEntries);
            if (recordsResult.getFailedRecordCount() == 0) {
                logger.debug("write {} tuple records ok", recordEntries.size());
                recordEntries.clear();//clear 操作只在全部写入后执行，若失败不清空
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }

    }

    /**
     * 写入Blob表记录
     *
     * @param recordData List of {@link BlobRecordData} 待写入的记录列表
     */
    public void writeBlob(List<BlobRecordData> recordData) {
        List<RecordEntry> recordEntries = new ArrayList<>();

        for (BlobRecordData recordDatum : recordData) {
            RecordEntry recordEntry = new RecordEntry();
            recordEntry.setRecordData(recordDatum);
            recordEntries.add(recordEntry);
        }
        try {
            PutRecordsResult recordsResult = commitRecordEntries(recordEntries);
            if (recordsResult.getFailedRecordCount() == 0) {
                logger.debug("write {} blob records ok", recordEntries.size());
                recordEntries.clear();//clear 操作只在全部写入后执行，若失败不清空
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * 带重试的提交记录
     *
     * @param recordEntries list of {@link RecordEntry} 待提交记录集合
     * @return {@link PutRecordsResult}
     * @throws SysException 提交失败 {@link PublishError} 提交记录失败
     */
    private PutRecordsResult commitRecordEntries(final List<RecordEntry> recordEntries) throws Exception {

        return RetryUtil.executeWithRetry(() -> {
            PutRecordsResult result = this.client.putRecords(this.projectName, this.topicName, recordEntries);
            if (result.getFailedRecordCount() > 0) {
                logger.error("写入数据失败,原计划写入{}条,实际写入失败{}条,即将进行重试.", recordEntries.size(), result.getFailedRecordCount());
                throw SysException.newException(PublishError.PUT_RECORD_ERROR);
            }
            return result;
        }, this.retryTimes, this.sleepTime, this.exponential);
    }


    public void setRetryTimes(int retryTimes) {
        this.retryTimes = retryTimes;
    }

    public void setSleepTime(long sleepTime) {
        this.sleepTime = sleepTime;
    }

    public void setExponential(boolean exponential) {
        this.exponential = exponential;
    }
}
